package com.example.capsule_netty.impl;


import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.example.capsule_netty.bean.*;
import com.example.capsule_netty.configuration.KafkaConstant;
import com.example.capsule_netty.global.GlobalThreadPool;
import com.example.capsule_netty.mapper.GroupMessageMapper;
import com.example.capsule_netty.mapper.OneToOneMessageMapper;
import com.example.capsule_netty.interfaces.MessageActionInterface;
import com.example.capsule_netty.mapper.UserInfoMapper;
import com.example.capsule_netty.mapper.UserToImeiBeanMapper;
import com.example.capsule_netty.global.UserAndChannelRelevance;
import com.example.capsule_netty.protocol.MessageBeanProtocol;
import com.google.gson.Gson;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.DefaultChannelGroup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Example;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhuzhiming
 * @Date: 2022/4/27 20:27
 */

@Component
@Slf4j
public class MessageActionImpl implements MessageActionInterface {

    @Autowired
    UserAndChannelRelevance userAndChannelRelevance;

    @Autowired
    UserToImeiBeanMapper userToImeiBeanMapper;

    @Resource
    OneToOneMessageMapper oneToOneMessageMapper;

    @Resource
    GroupMessageMapper groupMessageMapper;

    @Resource
    UserInfoMapper userInfoMapper;

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    RedisTemplate<String, String> redisTemplate;


    @Override
    public void dealRegisterAction(ChannelHandlerContext ctx, MessageBean messageBean) {

        String imei = messageBean.getImei();
        //String senderId = messageBean.getSenderId();
        List<UserToImeiBean> userList = userToImeiBeanMapper.getUserByMesi(imei);
        if (userList.size() == 0) {
            UserToImeiBean user = new UserToImeiBean();
            UserToImeiBean userToImeiBean = userToImeiBeanMapper.judgeHasOne();
            int maxId = 1;
            if (userToImeiBean == null) {
                userToImeiBeanMapper.doTruncateTable();
            } else {
                maxId = userToImeiBeanMapper.getMaxId();
            }
            user.setId(maxId == 1 ? 1 : maxId + 1);
            user.setUserId(String.valueOf(maxId == 1 ? 1 : maxId + 1));
            user.setImei(imei);
            userToImeiBeanMapper.save(user);
        }
        userAndChannelRelevance.putFromImei(imei, ctx.channel());
    }


    @Override
    public void dealConnectAction(ChannelHandlerContext ctx, MessageBean messageBean) {
        userAndChannelRelevance.putFromRoom(messageBean.getRoomName(), messageBean.getSenderId(), ctx.channel());
    }

    @Override
    public void dealChatAction(Channel channel, MessageBean messageBean) {

        if (messageBean.getMsgType() == 0 && !messageBean.isOneToOne()) {
            GlobalThreadPool.buildThreadPool().execute(new Runnable() {
                @Override
                public void run() {
                    kafkaTemplate.send(KafkaConstant.SAVE_MESSAGE_TOPIC, new Gson().toJson(messageBean)).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                        @Override
                        public void onFailure(Throwable ex) {
                            log.warn("send save message fail");
                        }

                        @Override
                        public void onSuccess(SendResult<String, String> result) {
                            //  log.info("send save message success");
                        }
                    });

                    kafkaTemplate.send(KafkaConstant.ES_DOC_INSERT_TOPIC, new Gson().toJson(messageBean)).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                        @Override
                        public void onFailure(Throwable ex) {
                            log.warn("send es doc insert message fail");
                        }

                        @Override
                        public void onSuccess(SendResult<String, String> result) {
                            //  log.info("send save message success");
                        }
                    });
                }
            });
        }

        if (messageBean.isOneToOne()) {

            if (preOperation(messageBean)) return;

            String receiverId = messageBean.getReceiverId();
            String imei = "";
            UserInfoBean userInfoBean;
            if (receiverId != null && !receiverId.equals("")) {

                String receiverInfo = redisTemplate.opsForValue().get(receiverId);

                if (messageBean.getContent() != null) {
                    System.out.println("私聊:  " + (messageBean.getContent().length() > 20 ? messageBean.getContent().substring(0, 20) : messageBean.getContent()));
                }

                if (!StringUtils.hasLength(receiverInfo)) {
                    userInfoBean = userInfoMapper.selectById(Integer.valueOf(receiverId));
                    redisTemplate.opsForValue().set(receiverId, new Gson().toJson(userInfoBean), 10, TimeUnit.MINUTES);
                } else {
                    userInfoBean = new Gson().fromJson(receiverInfo, UserInfoBean.class);
                }

                imei = userInfoBean.getImei();
                Channel receiveChannel = userAndChannelRelevance.getChannelFromImei(imei);

                if (receiveChannel == null) {
                    messageBean.setStatus(0);
                } else {
                    HashSet<Channel> channels = userAndChannelRelevance.getChannels(messageBean.getRoomName());
                    if (channels.size() == 0 || (channels.size() == 1 && channel == channels.iterator().next())) {
                        messageBean.setStatus(0);
                    } else {
                        messageBean.setStatus(1);
                    }
                }

                if (messageBean.getMsgType() < 7) {
                    CompletableFuture.runAsync(() -> {
                        kafkaTemplate.send(KafkaConstant.SAVE_MESSAGE_TOPIC, new Gson().toJson(messageBean));
                    });
                }

                sendToSelf(channel, messageBean);

                String json = new Gson().toJson(messageBean);
                byte[] jsonBytes = json.getBytes(StandardCharsets.UTF_8);
                MessageBeanProtocol messageBeanProtocol = new MessageBeanProtocol();
                messageBeanProtocol.setLen(json.getBytes(StandardCharsets.UTF_8).length);
                messageBeanProtocol.setContent(jsonBytes);
                if (receiveChannel != null || messageBean.getStatus() == 1) { //在线
                    receiveChannel.writeAndFlush(messageBeanProtocol);
                }

            }
        } else {

            if (preOperation(messageBean)) return;

            if (messageBean.getContent() != null) {
                System.out.println("群聊:  " + (messageBean.getContent().length() > 20 ? messageBean.getContent().substring(0, 20) : messageBean.getContent()));
            }
            String room = messageBean.getRoomName();
            // ByteBuf byteBuf = Unpooled.copiedBuffer(messageBean.getRoomName() + "|来自群聊的消息：   " + messageBean.getContent(), StandardCharsets.UTF_8);
            Set<Channel> channels = userAndChannelRelevance.getChannels(room);

            sendToSelf(channel, messageBean);

            for (Channel currChannel : channels) {
                if (currChannel != channel) {
                    String json = new Gson().toJson(messageBean);
                    byte[] jsonBytes = json.getBytes(StandardCharsets.UTF_8);
                    MessageBeanProtocol messageBeanProtocol = new MessageBeanProtocol();
                    messageBeanProtocol.setLen(json.getBytes(StandardCharsets.UTF_8).length);
                    messageBeanProtocol.setContent(jsonBytes);
                    currChannel.writeAndFlush(messageBeanProtocol);
                }
            }

        }

    }

    private void sendToSelf(Channel channel, MessageBean messageBean) {
        CompletableFuture.runAsync(() -> {
            MessageBean copyBean = new MessageBean();
            BeanUtils.copyProperties(messageBean, copyBean);
            copyBean.setContent(null);
            String json = new Gson().toJson(copyBean);
            byte[] jsonBytes = json.getBytes(StandardCharsets.UTF_8);
            MessageBeanProtocol messageBeanProtocol = new MessageBeanProtocol();
            messageBeanProtocol.setLen(json.getBytes(StandardCharsets.UTF_8).length);
            messageBeanProtocol.setContent(jsonBytes);
            channel.writeAndFlush(messageBeanProtocol);
        });
    }

    private boolean preOperation(MessageBean messageBean) {
        if (messageBean.getMsgType() == 7 && !messageBean.getContent().equals("")) {
            boolean needToSub = userAndChannelRelevance.removeByRoomAndUserUniqueMark(messageBean.getRoomName(), messageBean.getSenderId());
            if (!needToSub) {
                return true;
            }
        } else if (messageBean.getMsgType() == 8) {
            CompletableFuture.runAsync(() -> {
                if (!messageBean.isOneToOne()) {
                    groupMessageMapper.recallMessage(messageBean.getMsg().getId());
                } else {
                    oneToOneMessageMapper.recallMessage(messageBean.getMsg().getId());
                }
            });
        } else if (messageBean.getMsgType() < 7) {
            String maxId = redisTemplate.opsForValue().get(messageBean.getRoomName() + "maxId");
            int msgId;
            if (!StringUtils.hasLength(maxId)) {

                if (!messageBean.isOneToOne()) {
                    QueryWrapper<GroupMessageBean> queryWrapper = new QueryWrapper<GroupMessageBean>()
                            .eq("field", messageBean.getField())
                            .eq("room_name", messageBean.getRoomName())
                            .last("limit 1");
                    GroupMessageBean groupMessageBean = groupMessageMapper.selectOne(queryWrapper);
                    if (groupMessageBean == null) {
                        groupMessageMapper.truncateTable();
                        msgId = 1;
                    } else {
                        msgId = groupMessageMapper.getMaxId(messageBean.getField(), messageBean.getRoomName());
                    }
                } else {
                    QueryWrapper<OneToOneMessageBean> queryWrapper = new QueryWrapper<OneToOneMessageBean>()
                            .eq("room_name", messageBean.getRoomName())
                            .last("limit 1");
                    OneToOneMessageBean oneMessageBean = oneToOneMessageMapper.selectOne(queryWrapper);
                    if (oneMessageBean == null) {
                        oneToOneMessageMapper.truncateTable();
                        msgId = 1;
                    } else {
                        msgId = oneToOneMessageMapper.getMaxId(messageBean.getRoomName());
                    }
                }

            } else {
                msgId = Integer.parseInt(maxId);
            }
            messageBean.setMsgId(msgId);
            msgId++;
            redisTemplate.opsForValue().set(messageBean.getRoomName() + "maxId", msgId + "", 10, TimeUnit.MINUTES);
        }
        return false;
    }

    @Override
    public void dealSignAction(ChannelHandlerContext ctx, MessageBean messageBean) {

    }

    @Override
    public void dealClose(ChannelHandlerContext ctx, String imei) {
        CompletableFuture.runAsync(() -> {
            signFriendLeave(imei);
        });
        userAndChannelRelevance.removeByChannel(ctx.channel());
    }


    @Override
    public void dealExceptionAction(String imei) {
        userAndChannelRelevance.removeUserFromServer(imei);
    }


    private void signFriendLeave(String imei) {
        QueryWrapper<UserInfoBean> queryWrapper = new QueryWrapper<UserInfoBean>().eq("imei", imei);
        UserInfoBean userInfoBean = userInfoMapper.selectOne(queryWrapper);

        CompletableFuture.runAsync(() -> {
            userInfoBean.setOnlineStatus(0);
            String userRedisCache = redisTemplate.opsForValue().get(String.valueOf(userInfoBean.getId()));
            if (userRedisCache != null) {
                redisTemplate.opsForValue().set(String.valueOf(userInfoBean.getId()), new Gson().toJson(userInfoBean),10,TimeUnit.MINUTES);
            }
            userInfoMapper.updateById(userInfoBean);
        });

        String[] friends = userInfoBean.getFriends().split(",");

        if (friends.length > 20) {
            int step = friends.length / 20;
            for (int i = 0; i < 5; i++) {
                int finalI = i;
                GlobalThreadPool.buildThreadPool().execute(new Runnable() {
                    @Override
                    public void run() {
                        for (int j = finalI * step; j < (finalI == 4 ? friends.length : finalI * step + step); j++) {
                            sendLeaveMessage(friends[j], userInfoBean.getId());
                        }
                    }
                });
            }
        } else {
            for (String friendId : friends) {
                sendLeaveMessage(friendId, userInfoBean.getId());
            }
        }

    }


    private void sendLeaveMessage(String friendId, int userId) {
        UserInfoBean userInfoBean = userInfoMapper.selectById(friendId);
        String imei = userInfoBean.getImei();

        Channel channel = userAndChannelRelevance.getChannelFromImei(imei);

        if (channel == null) {
            return;
        }

        MessageBean messageBean = new MessageBean();
        messageBean.setContent("下线");
        messageBean.setStatus(0);
        messageBean.setMsgType(9);
        messageBean.setSenderId(String.valueOf(userId));
        messageBean.setOneToOne(true);
        String json = new Gson().toJson(messageBean);
        byte[] jsonBytes = json.getBytes(StandardCharsets.UTF_8);
        MessageBeanProtocol messageBeanProtocol = new MessageBeanProtocol();
        messageBeanProtocol.setLen(json.getBytes(StandardCharsets.UTF_8).length);
        messageBeanProtocol.setContent(jsonBytes);

        channel.writeAndFlush(messageBeanProtocol);
    }
}
